// share_state use : Arc<Mutex<Data>> between threads (tasks)
#[cfg(test)] 
mod test {
    //TODO: a example 
    // minimize the contention of mutex   TcpStream.splice & TcpStream.into_split();
    // decrease the mutex granular (restructure the code & shard the data)
    // use message passing  
    

    // read & write of client in two task ??? 
    // solution: [one task get (&mut client), another task set(&mut client)
    // std::sync::Arc<tokio::sync::Mutex<Client>>  => two task read & write with mutex
    // client is duplex communication :  no-need lock; --> use split
    #[tokio::test(flavor = "multi_thread")]
    #[ignore]
    async fn read_write_in_each_task(){
        use std::sync::Arc;
        use tokio::sync::Mutex;
        use mini_redis::client;
        let client = client::connect("127.0.0.1:6379").await.unwrap();
        let cli = Arc::new(Mutex::new(client));

        tokio::spawn({
            let cli = cli.clone();
            async move {
                let mut c = cli.lock().await; //note:  sync::Mutex; =>  v = mutex.lock().unwrap();
                let res = c.get("foo").await.unwrap();
                println!("{:?}", res);
            }
        });
        tokio::spawn( {
            let cli = cli.clone();
            async move {
                let mut c = cli.lock().await;
                let res = c.set("foo", "bar".into()).await.unwrap();
                println!("{:?}", res);
            }
        });
    }

    use bytes::Bytes;
    use std::collections::HashMap;
    use std::sync::{Arc, Mutex};
    type Db = Arc<Mutex<HashMap<String, Bytes>>>;

    #[tokio::test(flavor = "multi_thread")]
    #[ignore]
    async fn shared_db_between_tasks() {
        use tokio::net::TcpListener;

        let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();

        let db = Arc::new(Mutex::new(HashMap::new()));
        loop {
            let (socket, _) = listener.accept().await.unwrap();
            let db = db.clone();
            tokio::spawn(
                async move {
                    process(socket, db).await;
                }
            );
        }
    }
    use tokio::net::TcpStream;
    async fn process(socket: TcpStream, db: Db) {
        use mini_redis::{Connection, Frame};
        use mini_redis::Command::{self, Get, Set};

        let mut connection = Connection::new(socket);

        while let Some(frame) = connection.read_frame().await.unwrap() {
            let response = match Command::from_frame(frame).unwrap() {
                Set(cmd) => {   
                    let mut db = db.lock().unwrap();
                    db.insert(cmd.key().to_string(), cmd.value().clone());
                    Frame::Simple("OK".to_string())
                }
                Get(cmd) => {
                    let db = db.lock().unwrap();
                    if let Some(value) = db.get(cmd.key()) {
                        Frame::Bulk(value.clone())
                    } else {
                        Frame::Null
                    }
                }
                cmd => panic!("unimplemented {:?}", cmd),
            };

            connection.write_frame(&response).await.unwrap();
        }
    }
}
